package top.gcl.controller;

import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

//kafka生产者程序
@RestController
public class KafkaPrdCtrl {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;


//    //事务1
//    @GetMapping("/kafka4/{message}")
//    @Transactional(rollbackFor = RuntimeException.class)
//    public String sendToKafka2(@PathVariable("message") String msg){
//        //事务的支持
//        kafkaTemplate.send("userlog2",msg);
//        if("error".equals(msg)){
//            int i=1/0;
//        }
//        kafkaTemplate.send("userlog2",msg+"anthor");
//        return "send success!"+msg;
//    }
////事务二
//    @GetMapping("/kafka3/{message}")
//    public String sendMessage3(@PathVariable("message") final String msg) {
//        // 声明事务：后面报错消息不会发出去
//        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback(){
//            @Override
//            public Object doInOperations(KafkaOperations operations) {
//                operations.send("userlog2", msg);
//                if("error".equals(msg)){
//                    int i=1/0;//报错 让send()方法回滚
//                }
//                return null;
//            }
//        });
//        return "消息已发送";
//    }


//    //生产者同步发送消息
//    @GetMapping("/kafka5/{message}")
//    public String sendMessage5(@PathVariable("message") String msg) {
//        ListenableFuture<SendResult<String,Object>> future = kafkaTemplate.send("userlog", "key_test",msg);
//        try {
//            SendResult<String,Object> result = future.get();
//            String content = String.format("发送消息成功：%s - %s - %s",
//                    result.getRecordMetadata().topic(),
//                    result.getRecordMetadata().partition(),
//                    result.getRecordMetadata().offset());
//            System.out.println(content);
//        }catch (Throwable e){
//            e.printStackTrace();
//        }
//        return "消息已发送";
//    }


//    //回调
//    @GetMapping("/send/{msg}")
//    public String send(@PathVariable String msg){
//        kafkaTemplate.send("userlog", msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
//            @Override
//            public void onFailure(Throwable ex) {
//                System.out.println("发送消息失败：" + ex.getMessage());
//            }
//            @Override
//            public void onSuccess(SendResult<String, Object> result) {
//                String content = String.format("发送消息成功：%s - %s - %s",
//                        result.getRecordMetadata().topic(),
//                        result.getRecordMetadata().partition(),
//                        result.getRecordMetadata().offset());
//                System.out.println(content);
//            }
//        });
////向notice主题发送消息
//        String content=String.format("msg：%s", msg);
//        System.out.println(content);
//        return "消息已经发送";
//    }


//    @GetMapping("/send/{msg}")
//    public String send(@PathVariable String msg){
//        kafkaTemplate.send("userlog", msg);
//        String content=String.format("msg：%s", msg);
//        System.out.println(content);
//        return "消息已经发送";
//    }

    //通过Hash系统会给数据分配到指定的分区    key值换成hash值
    @GetMapping("/send/{msg}")
    public String send(@PathVariable String msg){
        kafkaTemplate.send("userlog","like", msg).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败：" + ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                String content = String.format("发送消息成功：%s - %s - %s",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                System.out.println(content);
            }
        });
//向notice主题发送消息
        String content=String.format("msg：%s", msg);
        System.out.println(content);
        return "消息已经发送";
    }
}
